package com.xsq.rpc.consumer.listener;

import com.xsq.rpc.common.model.RPCNettyClient;
import com.xsq.rpc.common.model.RpcMapper;
import com.xsq.rpc.consumer.rpc.RpcClient;
import io.netty.channel.EventLoopGroup;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.springframework.util.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * @ClassName PathChildrenListener
 * @Description TODO
 * @Author xsq
 * @Date 2020/6/10 15:06
 **/
@Slf4j
public class PathChildrenListener {

    public final static Map<String, RpcMapper> RPCMAPPERS = new ConcurrentHashMap<>();

    private CuratorFramework client;

    private static CountDownLatch watch = new CountDownLatch(1);

    public PathChildrenListener(CuratorFramework client) {
        this.client = client;
    }

    /**
     * 注册监听
     * TreeCache: 可以将指定的路径节点作为根节点（祖先节点），对其所有的子节点操作进行监听，
     * 呈现树形目录的监听，可以设置监听深度，最大监听深度为 int 类型的最大值。
     */
    public void zkWatch(String path) throws Exception {
        TreeCache treeCache = new TreeCache(client, path);

        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData eventData = event.getData();
                switch (event.getType()) {
                    case NODE_ADDED:
                        log.warn(path + "节点添加" + eventData.getPath() + "\t添加数据为：" + new String(eventData.getData()));
                        //触发集合保存节点信息,并且与对应的服务端节点建立RPC连接
                        connectionServerNode(eventData.getPath(), new String(eventData.getData()));
                        break;
                    case NODE_UPDATED:
                        log.warn(eventData.getPath() + "节点数据更新\t更新数据为：" + new String(eventData.getData()) + "\t版本为：" + eventData.getStat().getVersion());
                        break;
                    case NODE_REMOVED:
                        log.warn(eventData.getPath() + "节点被删除");
                        //触发集合删除信息,并且与对应的服务端断开RPC连接
                        cutServerNodeConnection(eventData.getPath());
                        break;
                    default:
                        break;
                }
            }
        });

        treeCache.start();
        watch.await();  //如果不执行 watch.countDown()，进程会一致阻塞在 watch.await()
    }

    /**
     * 触发集合删除信息,并且与对应的服务端断开RPC连接
     *
     * @param path
     */
    private void cutServerNodeConnection(String path) {
        RpcMapper rpcMapper = RPCMAPPERS.get(path);
        EventLoopGroup eventLoopGroup = rpcMapper.getrPCNettyClient().getEventLoopGroup();
        //断开对应的RPC连接
        eventLoopGroup.shutdownGracefully();
        //从连接信息中剔除断开的服务端连接
        RPCMAPPERS.remove(path);
    }

    /**
     * 跟服务端节点建立连接,并保存连接信息
     *
     * @param path
     * @param data
     */
    private void connectionServerNode(String path, String data) {
        if (!StringUtils.isEmpty(data)) {
            String serverUrl = data.split(":")[0];
            Integer port = Integer.valueOf(data.split(":")[1]);
            //初始化rpc调用客户端,建立客户端和服务端的连接
            RPCNettyClient rPCNettyClient = RpcClient.initClient(serverUrl, port);
            System.out.println("客户端RPC连接服务端fullPath：" + path + "---serverUrl：" + serverUrl + "----port：" + port + "--成功");
            RpcMapper rpcMapper = new RpcMapper();
            rpcMapper.setPath(path);
            rpcMapper.setServerUrl(serverUrl);
            rpcMapper.setPort(port);
            rpcMapper.setrPCNettyClient(rPCNettyClient);
            RPCMAPPERS.put(path, rpcMapper);
        }
    }

    /**
     * 获取节点内容
     *
     * @param client
     * @param path
     * @return
     */
    public String getZnodeValue(CuratorFramework client, String path) {
        //获取节点内容
        try {
            return new String(client.getData().forPath(path));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    // 关闭连接
    private void closeZnode() {
        if (client != null) {
            client.close();
        }
    }


}
